package com.oscar.online

import java.sql.DriverManager

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis


// 连接助手对象
object ConnHelper extends Serializable{
  // 懒变量定义，使用的时候才初始化
  lazy val jedis = new Jedis("flum", 6379)
  def createMysqlConnection()={
    Class.forName("com.mysql.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://localhost:3306/recommender","root","123456")
  }
}

// 定义样例类
case class BookRating( userId: Int, bookId: Int, score: Double, timestamp: Int )
case class MysqlConfig( url: String,user: String,password: String )
case class BookRecsOnMySQL(bookId: Int, recs: String)


object OnlineRecommender {
  // 定义常量
  val MYSQL_STREAM_RECS = "StreamRecs"
  val MYSQL_RATING = "Rating"
  val BOOK_RECS = "bookRecs"
  val MAX_USER_RATINGS_NUM = 20
  val MAX_SIM_BOOKS_NUM = 20


  def main(args: Array[String]): Unit = {
    //定义一些配置信息
    val config = Map(
      "spark.cores" -> "local[*]",
      "mysql.url" -> "jdbc:mysql://localhost:3306/recommender",
      "mysql.user" -> "root",
      "mysql.password" -> "123456",
      "kafka.topic" -> "recommender"
    )
    // 创建一个spark config
    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OnlineRecommender")
    // 创建spark session
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    import spark.implicits._
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc,Seconds(5))
    // 声明一个隐式的配置对象
    implicit val mysqlConfig =
      MysqlConfig(config.get("mysql.url").get,config.get("mysql.user").get,config.get("mysql.password").get)


    //TODO: 实时推荐主体代码
    // 广播图书相似度矩阵数据
    //装换成为 Map[Int, String]
     val simBooksMatrix: collection.Map[Int, String] = spark.read
      .format("jdbc")
      .option("url", mysqlConfig.url)
      .option("dbtable", BOOK_RECS)
      .option("user", mysqlConfig.user)
      .option("password", mysqlConfig.password)
      .load()
      .as[BookRecsOnMySQL]
      .rdd
      .map { item =>
        (item.bookId, item.recs)
      }.collectAsMap()
//     定义广播变量
    val simBooksMatrixBC: Broadcast[collection.Map[Int, String]] = sc.broadcast(simBooksMatrix)

    // 加载用户评分数据
    val ratingDS = spark.read
      .format("jdbc")
      .option("url", mysqlConfig.url)
      .option("dbtable", MYSQL_RATING)
      .option("user", mysqlConfig.user)
      .option("password", mysqlConfig.password)
      .load()
      .as[BookRating]
    // 获得用户已经评分过的图书Id
    val ratingExist = ratingDS.map(x=>(x.userId,x.bookId)).rdd
     // 1. 先使用collect方法,将此rdd收集起来
    // 2. 在Rdd的嵌套操作中调用此广播变量
    val RatingExist = sc.broadcast(ratingExist.collect()).value.toArray
    println("加载用户评分数据====》")
    RatingExist.take(10).foreach(println)


    //创建到Kafka的连接
    val kafkaParam = Map(
      "bootstrap.servers" -> "flum:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest"
//      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    // 创建一个DStream
    var kafkaStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream(ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe(Array(config("kafka.topic")),kafkaParam))


    // UID|MID|SCORE|TIMESTAMP
    // 产生评分流
    val ratingStream: DStream[(Int, Int, Double, Int)] = kafkaStream.map { msg =>
      var attr = msg.value().split("\\|")
      val str: String = msg.value()
      (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt)
    }



    // 核心实时推荐算法
    ratingStream.foreachRDD(rdds=>
      rdds.foreachPartition(rddsPartition=> {
        val connection = ConnHelper.createMysqlConnection()
        //-------------------------------------------------------------------------------------
        rddsPartition.foreach {
          case (userId, bookId, score, timestamp) =>
            println("rating data coming!>>>>>>>>>>>>>>>>>>"+"userId:"+userId+"bookId:"+bookId+"score:"+score+"timestamp"+timestamp)

            // 1. 从redis里取出当前用户的最近评分，保存成一个数组Array[(bookId, score)]
            val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATINGS_NUM, userId, ConnHelper.jedis)

            // 2. 从广播变量相似度矩阵中拿到当前图书的相似度列表，再从所有的相似图书中进行过滤,作为备选列表，保存成一个数组Array[bookId]
            val allSimBooks: Array[Int] = simBooksMatrixBC.value(bookId).split("\\|").map(x => x.split("\\:")(0).trim.toInt)
            val candidateBooks: Array[Int] = allSimBooks.filter(x => !RatingExist.contains((userId, x))).take(10)

            println("当前图书最相似的图书列表，作为备选列表===============>")
            candidateBooks.foreach(println)

            // 3. 计算每个备选图书的推荐优先级，得到当前用户的实时推荐列表，保存成 Array[(bookId, score)]
            val streamRecs: Array[(Int, Double)]  = computeBookScore(candidateBooks, userRecentlyRatings, simBooksMatrixBC.value)
            println("每个备选图书的推荐得分为===》")
            streamRecs.foreach(println)

            // 4. 把推荐列表保存到mysqldb
            try{
              var ss = new StringBuilder
              for (x <- streamRecs) {ss.append("|" + x._1 + ":" + x._2)}
              val recs = ss.toString()
              val querySql = "select * from information_schema.TABLES where TABLE_NAME = 'streamrecs'"
              val hasTable = connection.createStatement().executeQuery(querySql)
              println("表是否存在："+hasTable)
              if(hasTable!=null){
                val querysql = "select * from streamrecs where userid ="+userId
                val hasuser = connection.createStatement().executeQuery(querysql).next()
                println("hasuser = "+hasuser)
                if(hasuser){
                  println("用户存在："+userId+"推荐图书记得分为："+recs)
                  val insertSql2 = "update streamrecs set recs = '"+recs+"'where userid='"+userId+"'"
                  connection.createStatement().execute(insertSql2)
                  println("存入成功！！")
                  }else{
                  println("用户不存在："+userId+"推荐图书记得分为："+recs)
                  val insertSql = "INSERT INTO streamrecs(userId,recs) VALUES ("+userId+",'"+recs+"')"
                  connection.createStatement().execute(insertSql)
                  println("存入成功！！！")

                  }
              }else{
                val insertSql = "create table streamrecs(userId int(10) primary key auto_increment, recs text not null)"
                connection.createStatement().execute(insertSql)
                val createSql = "INSERT INTO streamrecs(userId,recs) VALUES ("+userId+",'"+recs+"')"
                connection.createStatement().execute(createSql)
              }
            }catch {
              case e: NullPointerException => println(e)
              println("抱歉，暂时没有图书推荐")
            }

        }
        //-------------------------------------------------------------------------------------
        connection.close()
      })
    )

    //启动Streaming程序
    ssc.start()
    ssc.awaitTermination()
  }
  //========================================================================================
  //从redis里获取最近num次评分
  import scala.collection.JavaConversions._
  def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis)
  : Array[(Int, Double)] = {
    // 从redis中用户的评分队列里获取评分数据，list键名为uid:USERID，值格式是 BOOKID:SCORE
    jedis.lrange( "userId:" + userId.toString, 0, num )
      .map{ item =>
        val attr = item.split("\\:")
        ( attr(0).trim.toInt, attr(1).trim.toDouble )
      }.toArray
  }


  //========================================================================================
  // 计算每个备选图书的推荐得分
  def computeBookScore(candidateBooks: Array[Int],
                          userRecentlyRatings: Array[(Int, Double)],
                          simBooks: scala.collection.Map[Int, scala.Predef.String])
  : Array[(Int, Double)] ={
      // 定义一个长度可变数组ArrayBuffer，用于保存每一个备选图书的基础得分，(bookId, score)
      val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
      // 定义两个map，用于保存每个图书的高分和低分的计数器，bookId -> count
      val increMap = scala.collection.mutable.HashMap[Int, Int]()
      val decreMap = scala.collection.mutable.HashMap[Int, Int]()

      // 遍历每个备选图书，计算和已评分图书的相似度
      for( candidateBook <- candidateBooks; userRecentlyRating <- userRecentlyRatings ){
          // 从相似度矩阵中获取当前备选图书和当前已评分图书间的相似度
          val simScore = getBooksSimScore( candidateBook, userRecentlyRating._1, simBooks )
          if( simScore > 0.35 ){
            // 按照公式进行加权计算，得到基础评分
            scores += ( (candidateBook, simScore * userRecentlyRating._2) )
            if( userRecentlyRating._2 > 3 ){
              increMap(candidateBook) = increMap.getOrDefault(candidateBook, 0) + 1
            } else {
              decreMap(candidateBook) = decreMap.getOrDefault(candidateBook, 0) + 1
            }
          }
      }

      // 根据公式计算所有的推荐优先级，首先以bookId做groupby
    scores.groupBy(_._1).map {
      case (bookId, scoreList) =>
        (bookId, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(bookId, 1)) - log(decreMap.getOrDefault(bookId, 1)))
    }.toArray // Map[Int, Double] -->Array[(Int, Double)]返回推荐列表，按照得分排序
      .sortWith(_._2 > _._2)
  }

  //========================================================================================
  //自定义getBooksSimScore函数
  def getBooksSimScore(book1: Int, book2: Int, simBooks:  scala.collection.Map[Int, scala.Predef.String]): Double ={
    val simBooksMap:scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]] =
      simBooks.map(x => (x._1,StringToMap(x._2)))
   simBooksMap.get(book1) match {
      case Some(sims) => sims.get(book2) match {
        case Some(score) => score
        case None => 0.0
      }
      case None => 0.0
    }
  }
  // 自定义log函数，以N为底
  def log(m: Int): Double = {
    val N = 10
    math.log(m)/math.log(N)
  }
  //自定义StringToMap函数
  def StringToMap(str: String): Map[Int, Double] = {
    str.split("\\|").map(x => {
      val attr=x.split("\\:")
      (attr(0).trim.toInt,attr(1).trim.toDouble)//Array[Array[Int, Double]]
    }).toMap
  }
}
